package com.atguigu.flink.edu.app.dwd.db;

import com.atguigu.flink.edu.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class dwd_trade_order_detail {

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(10L));
        tableEnv.executeSql(MyKafkaUtil.getTopicDbDDL("dwd_trade_order_detail"));
        Table orderInfo = tableEnv.sqlQuery("select \ndata['id'] order_id,\ndata['user_id'] user_id,\ndata['final_amount'] final_amount,\ndata['province_id'] province_id\nfrom `topic_db`\nwhere `table` = 'order_info'\nand `type` = 'insert'");
        tableEnv.createTemporaryView("order_info", orderInfo);
        Table orderDetail = tableEnv.sqlQuery("select \ndata['user_id'] user_id,\ndata['course_id'] course_id,\ndata['course_name'] course_name,\ndata['order_id'] order_id,\nts\nfrom `topic_db` where `table` = 'order_detail' and `type` = 'insert'\n");
        tableEnv.createTemporaryView("order_detail", orderDetail);
        Table resultTable = tableEnv.sqlQuery("select \noi.order_id,\noi.user_id,\noi.final_amount,\noi.province_id,\nod.course_id,\nod.course_name,\nts\nfrom order_detail od\ninner join order_info  oi\non od.order_id = oi.order_id");
        tableEnv.createTemporaryView("result_table", resultTable);
        tableEnv.executeSql("create table dwd_trade_order_detail_new(\norder_id string,\nuser_id string,\nfinal_amount string,\nprovince_id string,\ncourse_id string,\ncourse_name string,\nts string,\nprimary key (order_id) not enforced\n)" + MyKafkaUtil.getUpsertKafkaDDL("dwd_trade_order_detail_inc"));
        tableEnv.executeSql("insert into dwd_trade_order_detail_new select * from result_table");
        Table sqlQuery = tableEnv.sqlQuery("select * from dwd_trade_order_detail_new");
        tableEnv.createTemporaryView("order_new",sqlQuery);
        tableEnv.executeSql("select * from order_new").print();
    }
}
